package com.chencong.online.function;

import com.chencong.online.bean.ItemViewCountBean;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;

/**
 * @Author chencong
 * @Description 自定义函数：排序后取前五
 * 思路：将数据存在list状态里进行排序
 * @Date 8:00 下午 2021/12/19
 * @Param
 **/
public class TopNItemsCountProcessFunc extends KeyedProcessFunction<Tuple, ItemViewCountBean, String> {
    //定义topN大小
    private Integer topSize;

    //构造器
    public TopNItemsCountProcessFunc(Integer topSize) {
        this.topSize = topSize;
    }

    //定义listState状态，将保存当前窗口内所有输出的ItemViewCount
    ListState<ItemViewCountBean> itemViewCountListState;

    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化状态.从运行时下文中获取
        itemViewCountListState = getRuntimeContext().getListState(new ListStateDescriptor<ItemViewCountBean>("item-view-count-list-state", ItemViewCountBean.class));

    }

    @Override
    public void processElement(ItemViewCountBean value, Context ctx, Collector<String> out) throws Exception {
        //将状态存入
        itemViewCountListState.add(value);
        //注册定时器（同一时间的定时器注册多次都一样）
        ctx.timerService().registerEventTimeTimer(value.getWindowEnd() + 1);//加1ms的原因时为了触发定时器，因为窗口时左闭右开【 ）
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
        //当定时器触发时，收集窗口內的数据排序
        ArrayList<ItemViewCountBean> itemViewCounts = Lists.newArrayList(itemViewCountListState.get().iterator());
        //利用compare排序,倒序
        itemViewCounts.sort(new Comparator<ItemViewCountBean>() {
            @Override
            public int compare(ItemViewCountBean o1, ItemViewCountBean o2) {
                return o2.getCount().intValue() - o1.getCount().intValue();
            }
        });

        //为了直观查看list数据,打印输出，将他转为string查看，利用StringBuilder
        StringBuilder stringBuilder = new StringBuilder();

        stringBuilder.append("===================================\n");

        stringBuilder.append("窗口结束时间：").append(new Timestamp(timestamp - 1)).append("\n"); //刚刚定时器上面+1ms

        // 遍历列表，取top n输出
        for (int i = 0; i < Math.min(itemViewCounts.size(), topSize); i++) { //取之间的最小值，防止输出的topN太大，报空指针
            ItemViewCountBean currentItemViewCount = itemViewCounts.get(i);
            stringBuilder.append("NO ").append(i + 1).append(":")
                    .append("  商品ID  = ").append(currentItemViewCount.getItemId())
                    .append("  热门度  = ").append(currentItemViewCount.getCount())
                    .append("\n");
        }

        stringBuilder.append("================================\n\n");

        Thread.sleep(1000L);

        out.collect(stringBuilder.toString());

    }
}
